package com.sleep.cloudstreamdemo.consumer;

import com.sleep.cloudstreamdemo.domian.DemoMessage;
import lombok.extern.slf4j.Slf4j;
import org.springframework.context.annotation.Bean;
import org.springframework.integration.StaticMessageHeaderAccessor;
import org.springframework.integration.acks.AcknowledgmentCallback;
import org.springframework.messaging.Message;
import org.springframework.stereotype.Component;

import java.util.function.Consumer;

@Component
@Slf4j
public class LogCOnsumer {

    @Bean
    Consumer<Message<DemoMessage>> log(){
        return msg -> {

            AcknowledgmentCallback callback = StaticMessageHeaderAccessor.getAcknowledgmentCallback(msg);
            assert callback != null;
            try{
                log.debug("log日志消息：{}",msg.getPayload());
                // 消费者处理消息
                // 手动确认消息
                callback.acknowledge(AcknowledgmentCallback.Status.ACCEPT);
            }catch (Exception e){
                callback.acknowledge(AcknowledgmentCallback.Status.REJECT);
            }
            log.debug("log日志消息：{}",msg.getPayload());
        };
    }
}
